package WoidCount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.BZip2Codec;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.io.compress.bzip2.BZip2DummyCompressor;
import org.apache.hadoop.io.compress.bzip2.Bzip2Compressor;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 相当于一个yarn集群的客户端，
 * 需要在此封装我们的mr程序相关运行参数，指定jar包
 * 最后提交给yarn
 *
 * @author Administrator
 */
public class WordcountDriver {
    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();

        args = new String[]{"F:\\date\\A\\phone_num.txt", "F:\\date\\A\\Andymapyasuo1"};

        // 1 获取配置信息，或者job对象实例
        Configuration configuration = new Configuration();
        //开启map端压缩
        configuration.setBoolean("mapreduce.map.output.compress", true);
        //map端的压缩方式
        configuration.setClass("mapreduce.map.output.compress.codec", GzipCodec.class,CompressionCodec.class);

        Job job = Job.getInstance(configuration);

        job.setJarByClass(WordcountDriver.class);
        job.setMapperClass(WordcountMapper.class);
        job.setReducerClass(WordcountReducer.class);

        // 3 指定mapper输出数据的kv类型
        //Map输出的<K,V>  == Reduce输入的<K,V>
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //combiner
//        job.setCombinerClass(WordcountReducer.class);

        // 如果不设置InputFormat，它默认用的是
//        job.setInputFormatClass(CombineTextInputFormat.class);
//        CombineTextInputFormat.setMaxInputSplitSize(job, 3 * 1024 * 1024);
//        CombineTextInputFormat.setMinInputSplitSize(job, 2 * 1024 * 1024);

        //设置分区
//        job.setPartitionerClass(WordCountPartitioner.class);
//        job.setNumReduceTasks(7);

        // 5 指定job的输入原始文件所在目录
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //开启reduce端的压缩
        FileOutputFormat.setCompressOutput(job, true);
        //压缩方式
        FileOutputFormat.setOutputCompressorClass(job, BZip2Codec.class);


        // 7 将job中配置的相关参数，以及job所用的java类所在的jar包， 提交给yarn去运行
        job.waitForCompletion(true);


        long endTime = System.currentTimeMillis();

        System.out.println(endTime - startTime);
    }
}
